package com.winit.schedule.task.tracker.monitor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.zk.ZKService;
import com.winit.schedule.task.tracker.TaskTrackerContextImpl;
import com.winit.schedule.task.tracker.event.ConfigChangeEvent;
import com.winit.schedule.task.tracker.event.ConfigChangeEventType;

public class ZKServiceMonitor extends ZKService {

    private static transient final Logger  logger               = LoggerFactory.getLogger(ZKServiceMonitor.class);

    // 用来定时检查zk状态
    //private final ScheduledExecutorService disconnectedExecutor = Executors.newScheduledThreadPool(1);

    private TaskTrackerContextImpl         context;

    public ZKServiceMonitor(TaskTrackerContextImpl taskTrackerContextImpl) throws Exception{
        super("ZKServiceMonitor");
        this.context = taskTrackerContextImpl;
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
        this.addListener(new ClusterListener() {

            @SuppressWarnings("unchecked")
            @Override
            public void masterChanage(boolean isMaster, String masterAddress) {
                getContext().getDispatcher()
                    .getEventHandler()
                    .handle(new ConfigChangeEvent(ConfigChangeEventType.MASTER_CHANGE, masterAddress));
            }

            @Override
            public void disconnected() {
            }

            @Override
            public void connected() {
            }
        });
    }

    @Override
    protected void serviceStart() throws Exception {
        // zookeeper断开连接，只要jobtracker能正常访问则tasktracker继续运行
        //disconnectedExecutor.scheduleWithFixedDelay(new DisconnectedThread(this), 3000, 5000, TimeUnit.MILLISECONDS);
        super.serviceStart();
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
    }

    public TaskTrackerContextImpl getContext() {
        return context;
    }

    class DisconnectedThread implements Runnable {

        ZKServiceMonitor zkServiceMonitor;

        public DisconnectedThread(ZKServiceMonitor zkServiceMonitor){
            this.zkServiceMonitor = zkServiceMonitor;
        }

        @SuppressWarnings("static-access")
        @Override
        public void run() {
            try {
                Thread.currentThread().setPriority(Thread.MAX_PRIORITY);

                while (true) {
                    Thread.currentThread().sleep(5000);

                    if (!zkServiceMonitor.checkZKState()) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("zookeeper is disconnected");
                        }
                        getContext().getTaskTracker().stopAllTask();
                    }
                }

            } catch (Exception ex) {
                logger.error("Check task expire fialure.", ex);
            }
        }

    }

}
